/*
    网络具象层：对抽象层缓冲区基类、协议基类、连接基类、服务端基类、客户端基类的派生，并实现每个派生类的工厂类
*/
#pragma once
#include <string>
#include <unordered_map>
#include <muduo/net/TcpServer.h>
#include <muduo/net/TcpClient.h>
#include <muduo/net/EventLoop.h>
#include <muduo/net/EventLoopThread.h>
#include <muduo/base/CountDownLatch.h>
#include <muduo/net/TcpConnection.h>
#include <muduo/net/Buffer.h>
#include "message.hpp"
#include <thread>
#include <sys/syscall.h> // for syscall and SYS_gettid
#include <unistd.h>      // for getpid

namespace ns_jsonrpc
{
    // 缓存类
    class MuduoBuffer : public BaseBuffer
    {
    public:
        using MuduoBufferPtr = std::shared_ptr<MuduoBuffer>;

    public:
        MuduoBuffer(muduo::net::Buffer *buf) : _buf(buf)
        {
        }

        /*
            注意：从muduo库的缓冲区读取的数据默认是从网络序转换为主机序的
        */
        // 获取缓冲区可读字节数
        virtual size_t readableBytes() override
        {
            return _buf->readableBytes();
        }

        // 查看缓冲区前4个字节，不删除数据
        virtual int32_t peekInt32() override
        {
            return _buf->peekInt32();
        }

        // 删除缓冲区前4个字节（缓冲区指针向后移动4字节）
        virtual void retrieveInt32() override
        {
            _buf->retrieveInt32();
        }

        // 读取缓冲区前前4个字节，并删除数据
        virtual int32_t readInt32() override
        {
            return _buf->readInt32();
        }

        // 读取缓冲区指定长度的字符串
        virtual std::string retrieveAsString(size_t len) override
        {
            return _buf->retrieveAsString(len);
        }

    private:
        muduo::net::Buffer *_buf;
    };

    // 生产MuduoBuffer的工厂类
    class MuduoBufferFactory
    {
    public:
        template <typename... Args>
        static BaseBuffer::BaseBufferPtr create(Args &&...args)
        {
            return std::make_shared<MuduoBuffer>(std::forward<Args>(args)...);
        }
    };

    // 应用层LV格式协议类
    class LVProtocol : public BaseProtocol
    {
    public:
        using LVProtocolPtr = std::shared_ptr<LVProtocol>;

    public:
        // 是否可以处理缓冲区中的请求（缓冲区中是否有有一条完成的请求）
        virtual bool canProcessed(const BaseBuffer::BaseBufferPtr &buf)
        {
            // |--Len--|--VALUE--|
            // |--Len--|--mtype--|--idlen--|--id--|--body--|
            if (buf->readableBytes() < lenFieldsLength) // 检查缓冲区有没有完成的len字段
                return false;
            size_t len = buf->peekInt32();
            return buf->readableBytes() >= len + lenFieldsLength;
        }

        // 收到消息时处理函数（调用该函数则默认缓冲区有完整的一条消息）
        virtual bool onMessage(const BaseBuffer::BaseBufferPtr &buf, BaseMessage::BaseMessagePtr &msg)
        {
            // 从缓冲区中安装LV格式读取数据，将数据存入到mtype格式的消息响应结构体中（msg）
            // |--Len--|--mtype--|--idlen--|--id--|--body--|
            size_t len = buf->readInt32();                                                                 // 读取消息内容总长度
            MType mtype = (MType)buf->readInt32();                                                         // 读取消息类型
            size_t idlen = buf->readInt32();                                                               // 读取消息id长度
            std::string id = buf->retrieveAsString(idlen);                                                 // 读取id内容
            std::string body = buf->retrieveAsString(len - mtypeFieldsLength - idlenFieldsLength - idlen); // 读取body内容
            msg = MessageFactory::create(mtype);
            if (msg.get() == nullptr)
            {
                LOG_ERROR("message type error, fail to create message\n");
                return false;
            }
            bool res = msg->unSerialize(body);
            if (!res)
            {
                LOG_ERROR("message unserialize error\n");
                return false;
            }
            msg->setId(id);       // 设置消息id
            msg->setMType(mtype); // 设置消息类型
            return true;
        }

        // 将消息序列化为标准协议格式
        virtual std::string serialize(const BaseMessage::BaseMessagePtr &msg)
        {
            // |--Len--|--mtype--|--idlen--|--id--|--body--|
            // 从消息中获取/计算LV协议中每个字段的值
            std::string body = msg->serialize();
            int32_t mtype = htonl((int32_t)msg->mType());
            std::string id = msg->id();
            int32_t idlen = htonl(id.size());
            int32_t totalLen = mtypeFieldsLength + idlenFieldsLength + id.size() + body.size();
            int32_t len = htonl(totalLen);
            
            // 拼接LV协议字符串
            std::string protoStr;
            protoStr.reserve(totalLen);
            protoStr.append((char *)&len, lenFieldsLength);
            protoStr.append((char *)&mtype, mtypeFieldsLength);
            protoStr.append((char *)&idlen, idlenFieldsLength);
            protoStr.append(id);
            protoStr.append(body);
            return protoStr;
        }

    private:
        const size_t lenFieldsLength = 4;
        const size_t mtypeFieldsLength = 4;
        const size_t idlenFieldsLength = 4;
    };

    // 生产LVProtocol的工厂类
    class LVProtocolFactory
    {
    public:
        template <typename... Args>
        static BaseProtocol::BaseProtocolPtr create(Args &&...args)
        {
            return std::make_shared<LVProtocol>(std::forward<Args>(args)...);
        }
    };

    // 连接抽象类
    class MuduoConnection : public BaseConnection
    {
    public:
        using MuduoConnectionPtr = std::shared_ptr<MuduoConnection>;

    public:
        // 注意：连接类主要负责连接管理、数据的发送（将消息发送出去，所以需要应用层协议类来将消息序列化，但是所有的连接只需要共享一个应用层协议类即可）
        MuduoConnection(const BaseProtocol::BaseProtocolPtr &protocol, const muduo::net::TcpConnectionPtr &conn) : _protocol(protocol), _conn(conn)
        {
        }

        // 获取连接状态
        virtual bool connected()
        {
            return _conn->connected();
        }

        // 关闭连接
        virtual void shutdown()
        {
            _conn->shutdown();
        }

        // 发送消息
        virtual void send(const BaseMessage::BaseMessagePtr &msg)
        {
            std::string protoStr = _protocol->serialize(msg);
            _conn->send(protoStr);
        }

    private:
        BaseProtocol::BaseProtocolPtr _protocol;
        muduo::net::TcpConnectionPtr _conn;
    };

    // 生产MuduoConnection的工厂类
    class MuduoConnectionFactory
    {
    public:
        template <typename... Args>
        static BaseConnection::BaseConnectionPtr create(Args &&...args)
        {
            return std::make_shared<MuduoConnection>(std::forward<Args>(args)...);
        }
    };

    // 服务器抽象类
    class MuduoServer : public BaseServer
    {
    public:
        using MuduoServerPtr = std::shared_ptr<MuduoServer>;
        /*
            整体通信的管理
        */
    public:
        MuduoServer(int port = 8888) : _server(&_baseloop, muduo::net::InetAddress("0.0.0.0", port), "MuduoServer", muduo::net::TcpServer::kReusePort),
                                       _protocol(LVProtocolFactory::create())
        {
        }

        // 启动服务
        virtual void start()
        {
            // 设置连接建立/断开回调、消息处理回调
            _server.setConnectionCallback(std::bind(&MuduoServer::onConnection, this, std::placeholders::_1));
            _server.setMessageCallback(std::bind(&MuduoServer::onMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

            // 开始监听
            _server.start();
            LOG_INFO("server start......\n");

            // 开始事件循环
            _baseloop.loop();
        }

    private:
        // 连接建立或断开时的处理函数
        void onConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected())
            {
                // 连接建立
                LOG_DEBUG("onConnection: %s -> %s\n", conn->peerAddress().toIpPort().c_str(), conn->localAddress().toIpPort().c_str());
                std::shared_ptr<BaseConnection> muduoConn = MuduoConnectionFactory::create(_protocol, conn);
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    _conns[conn] = muduoConn;
                }
                if (_cb_connection)
                    _cb_connection(muduoConn);
            }
            else
            {
                // 连接断开
                LOG_DEBUG("onConnection: %s -> %s is down\n", conn->peerAddress().toIpPort().c_str(), conn->localAddress().toIpPort().c_str());
                BaseConnection::BaseConnectionPtr muduoConn;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _conns.find(conn);
                    if (it == _conns.end())
                        return;
                    muduoConn = it->second;
                    _conns.erase(it);
                }
                if (_cb_close)
                    _cb_close(muduoConn);
            }
        }

        // 连接读写事件处理函数
        void onMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buff, muduo::Timestamp)
        {
            BaseConnection::BaseConnectionPtr muduoConn; // muduo连接（用来发送消息）
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _conns.find(conn);
                if (it == _conns.end())
                {
                    LOG_ERROR("connection not found\n");
                    return;
                }
                muduoConn = it->second;
            }
            BaseBuffer::BaseBufferPtr muduoBuff = MuduoBufferFactory::create(buff); // muduo缓冲区（用来存储消息）
            while (1)
            {
                // 1.消息检查（判断缓冲区中是否至少有一条可以处理的消息）
                if (_protocol->canProcessed(muduoBuff) == false)
                {
                    // 恶意排查：len过大而导致无法处理
                    if (muduoBuff->readableBytes() > _max_data_size)
                    {
                        conn->shutdown();
                        LOG_ERROR("buffer data too long\n");
                        return;
                    }
                    // 消息量不够，等待下次
                    LOG_DEBUG("buffer data not enough, wait for the next time\n");
                    break;
                }

                // 2.获取消息
                BaseMessage::BaseMessagePtr msg;
                bool ret = _protocol->onMessage(muduoBuff, msg); // 从缓冲区中解析消息到对应消息类型结构体中（协议的反序列化）
                if (!ret)
                {
                    conn->shutdown();
                    LOG_ERROR("buffer data error\n");
                    return;
                }

                // 3.处理消息（一次处理一条消息）
                LOG_DEBUG("invoke callback function to process message\n");
                if (_cb_message)
                    _cb_message(muduoConn, msg);
            }
        }

    private:
        BaseProtocol::BaseProtocolPtr _protocol;
        muduo::net::EventLoop _baseloop;
        muduo::net::TcpServer _server;
        std::unordered_map<muduo::net::TcpConnectionPtr, BaseConnection::BaseConnectionPtr> _conns;
        std::mutex _mutex;
        const int _max_data_size = (1 << 16); // 64KB
    };

    // 生产MuduoServer的工厂类
    class MuduoServerFactory
    {
    public:
        template <typename... Args>
        static BaseServer::BaseServerPtr create(Args &&...args)
        {
            return std::make_shared<MuduoServer>(std::forward<Args>(args)...);
        }
    };

    // 客户端抽象类
    class MuduoClient : public BaseClient
    {
    public:
        using MuduoClientPtr = std::shared_ptr<MuduoClient>;

    public:
        MuduoClient(const std::string &sip, int sport) : _baseloop(_loopthread.startLoop()),
                                                         _downlatch(1),
                                                         _client(_baseloop, muduo::net::InetAddress(sip, sport), "MuduoClient"),
                                                         _protocol(LVProtocolFactory::create())
        {
        }

        // 连接
        virtual void connect() override
        {
            _client.setConnectionCallback(std::bind(&MuduoClient::onConnection, this, std::placeholders::_1));
            _client.setMessageCallback(std::bind(&MuduoClient::onMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _client.connect();
            _downlatch.wait();
        }

        // 关闭连接
        virtual void shutdown() override
        {
            _client.disconnect();
        }

        // 获取连接
        virtual BaseConnection::BaseConnectionPtr connection() override
        {
            return _conn;
        }

        // 获取连接状态
        virtual bool connected()
        {
            return (_conn && _conn->connected());
        }

        // 发送消息
        virtual bool send(const BaseMessage::BaseMessagePtr &msg) override
        {
            if (connected() == false)
            {
                LOG_ERROR("client not connected\n");
                return false;
            }
            _conn->send(msg);
            return true;
        }

    private:
        void onConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected())
            {
                LOG_DEBUG("client connected\n");
                _conn = MuduoConnectionFactory::create(_protocol, conn);
                _downlatch.countDown();
            }
            else
            {
                LOG_DEBUG("client disconnected\n");
                _conn.reset();
            }
        }

        void onMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buf, muduo::Timestamp time)
        {
            BaseBuffer::BaseBufferPtr muduoBuff = MuduoBufferFactory::create(buf);
            while (1)
            {
                // 1.消息检查（判断缓冲区中是否至少有一条可以处理的消息）
                if (_protocol->canProcessed(muduoBuff) == false)
                {
                    // 恶意排查：len过大而导致无法处理
                    if (muduoBuff->readableBytes() > _max_data_size)
                    {
                        conn->shutdown();
                        LOG_ERROR("buffer data too long\n");
                        return;
                    }
                    // 消息量不够，等待下次
                    LOG_DEBUG("buffer data not enough, wait for the next time\n");
                    break;
                }

                // 2.获取消息
                BaseMessage::BaseMessagePtr msg;
                bool ret = _protocol->onMessage(muduoBuff, msg); // 从缓冲区中解析消息到对应消息类型结构体中（协议的反序列化）
                if (!ret)
                {
                    conn->shutdown();
                    LOG_ERROR("buffer data error\n");
                    return;
                }

                // 3.处理消息（一次处理一条消息）
                LOG_DEBUG("invoke callback function to process message\n");
                if (_cb_message)
                    _cb_message(_conn, msg);
            }
        }

    private:
        muduo::net::EventLoopThread _loopthread;
        muduo::net::EventLoop *_baseloop;
        muduo::net::TcpClient _client;
        muduo::CountDownLatch _downlatch;
        BaseProtocol::BaseProtocolPtr _protocol;
        BaseConnection::BaseConnectionPtr _conn;
        const int _max_data_size = (1 << 16);
    };

    // 生产MuduoClient的工厂类
    class MuduoClientFactory
    {
    public:
        template <typename... Args>
        static BaseClient::BaseClientPtr create(Args &&...args)
        {
            return std::make_shared<MuduoClient>(std::forward<Args>(args)...);
        }
    };
}